// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/kraken/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/kraken/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"
#include "coin2/exchange/kraken/symbology/product.h"

namespace coin2::exchange::kraken::feed_v1::impl {

void KrakenFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(std::string("JSON parse error: ") +
                             GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.IsArray()) {
    auto items = document.GetArray();
    CHECK_THROW(items.Size() == 4 || items.Size() == 5)
        << "[Kraken Feed Processor] unexpected length msg: " << packet.payload;

    const auto& channel = (items.Size() == 4)
                              ? KrakenChannel::Parse(items[2].GetString())
                              : KrakenChannel::Parse(items[3].GetString());
    const auto& native_symbol =
        (items.Size() == 4) ? items[3].GetString() : items[4].GetString();

    if (!CheckChannelFilter(channel.type)) {
      return;
    }

    if (channel.type == "book") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        bool book_generated = KrakenFeedUpdateParser::ParseBook(book, items[1]);
        if (items.Size() == 5) {
          book_generated = KrakenFeedUpdateParser::ParseBook(book, items[2]);
        }
        if (book_generated) {
          upd_mgr_.PublishBook();
        }
      }
    } else if (channel.type == "trade") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }

      Trade* agg_trade = upd_mgr_.GetTradeMutable();
      agg_trade->Clear();
      Trade trade{};

      const rapidjson::Value& trades = items[1];
      agg_trade->timestamp =
          std::llround(stod(trades.GetArray()[0].GetArray()[2].GetString()) *
                       1'000'000'000LL);

      for (const auto& trade_json : trades.GetArray()) {
        KrakenFeedUpdateParser::ParseTrade(trade_json, &trade);
        if (TryAggregateTrades(agg_trade, trade)) {
          continue;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, true);
        }
        *agg_trade = trade;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else {
      LOG(INFO) << "[Kraken] unknown msg for " << packet.topic_name << ": "
                << packet.payload.data();
    }
  } else if (document.HasMember("url")) {
    const std::string url = document["url"].GetString();
    if (url.find("OHLC?")) {
      // kline data
      auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      const std::string& interval_str = parameters["interval"];
      if (!CheckChannelFilter("pair={}&interval=" + interval_str)) {
        return;
      }
      if (!document["data"].HasMember("result")) {
        LOG(INFO) << "[Kraken] kline query failed " << packet.topic_name << ": "
                << packet.payload.data();
      }

      // rest native symbol to websocket native symbol
      // XXBTZUSD -> XXBT/ZUSD
      rapidjson::Value::ConstMemberIterator iter;
      const auto& result = document["data"]["result"];
      CHECK_EQ(result.GetObject().MemberCount(), 2);
      for (iter = result.MemberBegin(); iter != result.MemberEnd(); iter++) {
        if (iter->name.GetString() != std::string("last")) {
          break;
        }
      }
      std::string rest_native_symbol = iter->name.GetString();
      std::string native_symbol = symbology::KrakenProduct::FromNativeStr(
          rest_native_symbol, packet.timestamp).feed_native_symbol();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          return;
      }

      auto interval = KrakenFeedUpdateParser::ParseKlineInterval(interval_str);
      auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (kline != nullptr) {

        const auto& kline_array = iter->value.GetArray();
        if (kline_array.Size() > 10) {
          // first long query
          kline->SnapshotClear();
        } else {
          kline->UpdateClear();
        }

        bool to_publish = false;
        for (const auto& kline_data : kline_array) {
          bool freezed = KrakenFeedUpdateParser::Parse1Kline(
              kline_data, kline, packet.timestamp, interval);
          to_publish = to_publish || freezed;
        }

        if (to_publish) {
          upd_mgr_.PublishKline(native_symbol);
        }
      }
    } else {
      LOG(INFO) << "[Kraken] unknown rest msg for " << packet.topic_name << ": "
                << packet.payload.data();
    }
  } else if (!(document.HasMember("event") &&
               strcmp(document["event"].GetString(), "heartbeat") == 0)) {
    // meta msg like subscription response...
    LOG(INFO) << "[Kraken Feed Processor] meta msg for " << packet.topic_name
              << ": " << packet.payload.data();
  }
  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::kraken::feed_v1::impl
